這一章來分享如何製作自己遊戲內的MCP
讓 AI 可以控製你的遊戲角色, 進行各種操作

要讓 AI 控制你的遊戲, 我們需要在 Unity 中創建一個伺服器,
然後讓 Python 調用裡面的代碼, AI 再聯動到這個 Python 的代碼去查看有什麼工具可以使用
我們只需要兩個簡單的代碼:
完整的專案源代碼放在Github: https://github.com/yayapipi/InGameMCP
simpleMCP.cs
https://github.com/yayapipi/InGameMCP/blob/main/Assets/SimpleMCP/SimpleMCP.cs
主要是會在 Unity 啟動的時候開啟一個 Server, 設定 Port:
private void StartServer()
{
    try
    {
        listener = new TcpListener(IPAddress.Any, port);
        listener.Start();
        isRunning = true;
        serverThread = new Thread(AcceptClients);
        serverThread.IsBackground = true;
        serverThread.Start();
        Debug.Log($"[SimpleMCP] 伺服器已啟動在端口 {port}");
        Debug.Log("[SimpleMCP] 建議透過 MCP Bridge 連接(Cursor 將依 .cursor/mcp.json 自動啟動 Python Bridge)");
    }
    catch (Exception e)
    {
        Debug.LogError($"[SimpleMCP] 啟動失敗: {e.Message}");
    }
}
接著就會不斷的監聽這個 Port 有沒有新的訊息進來,
當檢測到有人連進來的時候, 就會跟它握手, 然後建立長連線 WebSocket
開始不斷地從他那邊接受指令:
 private void HandleClient(TcpClient client)
    {
        NetworkStream stream = client.GetStream();
        byte[] buffer = new byte[4096];
        try
        {
            // WebSocket 握手
            int bytesRead = stream.Read(buffer, 0, buffer.Length);
            string request = Encoding.UTF8.GetString(buffer, 0, bytesRead);
            if (request.Contains("Upgrade: websocket"))
            {
                string response = PerformHandshake(request);
                byte[] responseBytes = Encoding.UTF8.GetBytes(response);
                stream.Write(responseBytes, 0, responseBytes.Length);
                Debug.Log("[SimpleMCP] 客戶端已連接");
                // 處理消息
                while (isRunning && client.Connected)
                {
                    if (stream.DataAvailable)
                    {
                        bytesRead = stream.Read(buffer, 0, buffer.Length);
                        if (bytesRead <= 0)
                        {
                            break; // 連線關閉
                        }
                        if (bytesRead > 0)
                        {
                            string message = DecodeFrame(buffer, bytesRead);
                            if (!string.IsNullOrEmpty(message))
                            {
                                ProcessMessage(message, stream);
                            }
                        }
                    }
                    Thread.Sleep(10);
                }
            }
        }
        catch (Exception e)
        {
            Debug.LogError($"[SimpleMCP] 客戶端錯誤: {e.Message}");
        }
        finally
        {
            client.Close();
            Debug.Log("[SimpleMCP] 客戶端已斷開");
        }
    }
當收到新的指令的時候, 會在 ProcessMessage/HandleRequest 裡面進行判斷:
這裡就是收到什麼指令, 就觸發什麼 Function, 也可以帶一些參數進來
執行完之後, 就把結果回傳回去
private void ProcessMessage(string message, NetworkStream stream)
    {
        lock (mainThreadActions)
        {
            mainThreadActions.Enqueue(() =>
            {
                try
                {
                    Debug.Log($"[SimpleMCP] 收到消息: {message}");
                    var request = JsonUtility.FromJson<MCPRequest>(message);
                    if (request == null || string.IsNullOrEmpty(request.method))
                    {
                        Debug.LogError("[SimpleMCP] 無效的請求格式");
                        SendResponse(stream, "{\"error\":\"Invalid request format\"}");
                        return;
                    }
                    string response = HandleRequest(request);
                    SendResponse(stream, response);
                }
                catch (Exception e)
                {
                    Debug.LogError($"[SimpleMCP] 處理消息錯誤: {e.Message}\n消息內容: {message}");
                    SendResponse(stream, "{\"error\":\"Internal server error\"}");
                }
            });
        }
    }
    // 新增的 HandleRequest 方法
    private string HandleRequest(MCPRequest request)
    {
        try
        {
            Debug.Log($"[SimpleMCP] 處理請求方法: {request.method}");
            switch (request.method.ToLower())
            {
                case "get_position":
                case "getposition":
                    return GetPlayerPosition();
                case "move_player":
                case "moveplayer":
                case "set_position":
                case "setposition":
                    return MovePlayer(request.x, request.y, request.z);
                case "ping":
                    return "{\"success\":true,\"message\":\"pong\"}";
                case "get_player_info":
                case "getplayerinfo":
                    return GetPlayerInfo();
                default:
                    Debug.LogWarning($"[SimpleMCP] 未知的請求方法: {request.method}");
                    return "{\"error\":\"Unknown method\",\"method\":\"" + request.method + "\"}";
            }
        }
        catch (Exception e)
        {
            Debug.LogError($"[SimpleMCP] HandleRequest 錯誤: {e.Message}");
            return "{\"error\":\"Request handling failed\",\"details\":\"" + e.Message + "\"}";
        }
    }
現在有兩個功能
知道這個原理之後, 我們就可以繼續在 HandleRequest 繼續擴充新的功能

代碼寫好之後,我們可以創建一個簡單的場景
把玩家 (Capsule) 放進去, 執行遊戲進行測試, 看會不會打開 Server
simple_mcp_bridge.py
https://github.com/yayapipi/InGameMCP/blob/main/Assets/SimpleMCP/PythonServer/simple_mcp_bridge.py
接下來我們需要一個 Python 的橋接器代碼
讓 AI 可以透過這個代碼跟我們的遊戲進行溝通
一開始啟動的時候需要先安裝 Websockets
pip3 install websockets
這個代碼一開始會想連接到 Unity WebSocket 的伺服器
async def connect_to_unity():
    """連接到 Unity WebSocket 伺服器"""
    global ws_connection
    
    log(f"Attempting to connect to Unity at {UNITY_URI}")
    
    max_retries = 5
    retry_delay = 2
    
    for attempt in range(max_retries):
        try:
            ws_connection = await asyncio.wait_for(
                websockets.connect(UNITY_URI),
                timeout=5.0
            )
            log(f"✓ Connected to Unity at {UNITY_URI}")
            return True
        except asyncio.TimeoutError:
            log(f"✗ Connection timeout (attempt {attempt + 1}/{max_retries})")
        except ConnectionRefusedError:
            log(f"✗ Connection refused - Is Unity running? (attempt {attempt + 1}/{max_retries})")
        except Exception as e:
            log(f"✗ Connection error: {type(e).__name__} - {e} (attempt {attempt + 1}/{max_retries})")
        
        if attempt < max_retries - 1:
            log(f"Retrying in {retry_delay} seconds...")
            await asyncio.sleep(retry_delay)
    
    log("✗ Failed to connect to Unity after all retries")
    log("Please ensure:")
    log("  1. Unity is running")
    log("  2. SimpleMCP script is attached to a GameObject")
    log("  3. The game is in Play mode")
    log(f"  4. Port {UNITY_PORT} is not blocked by firewall")
收到 AI 的指令的時候就給我們的遊戲發送指令:
async def send_unity_request(method, **kwargs):
    """發送請求到 Unity"""
    global ws_connection
    
    if ws_connection is None:
        log("WebSocket not connected, attempting to connect...")
        if not await connect_to_unity():
            return {"error": "Not connected to Unity. Is the game running?"}
    
    try:
        request = {"method": method, **kwargs}
        log(f"Sending to Unity: {json.dumps(request)}")
        
        await ws_connection.send(json.dumps(request))
        response = await asyncio.wait_for(ws_connection.recv(), timeout=5.0)
        
        log(f"Received from Unity: {response}")
        return json.loads(response)
        
    except asyncio.TimeoutError:
        log("✗ Unity request timeout")
        return {"error": "Unity request timeout"}
    except websockets.exceptions.ConnectionClosed as e:
        log(f"✗ Connection to Unity lost: {e}")
        ws_connection = None
        return {"error": "Connection to Unity lost"}
    except Exception as e:
        log(f"✗ Error sending request: {type(e).__name__} - {e}")
        return {"error": str(e)}
那麼 AI 是怎麼跟這個 Python 代碼進行溝通的呢?
關鍵的代碼在 handle_mcp_request 這個 function 裡:
一開始的時候 AI 會先進行初始化 Initialize
這個時候要回傳一些相關資訊給 AI, 讓他知道這個 MCP 是幹嘛的
# MCP 初始化握手
if method == "initialize":
    log("Handling initialize request")
    return {
        "protocolVersion": "2024-11-05",
        "capabilities": {
            "tools": {}
        },
        "serverInfo": {
            "name": "unity-game-mcp",
            "version": "1.0.0"
        }
    }
接著, 我們會在 tools/list 定義可以使用的工具有哪些
像我們目前有的指令是
elif method == "tools/list":
    return {
        "tools": [
            {
                "name": "get_player_position",
                "description": "獲取玩家在遊戲中的當前位置座標",
                "inputSchema": {
                    "type": "object",
                    "properties": {},
                    "required": []
                }
            },
            {
                "name": "move_player",
                "description": "將玩家移動到指定的座標位置",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "x": {"type": "number", "description": "X 座標"},
                        "y": {"type": "number", "description": "Y 座標(高度)"},
                        "z": {"type": "number", "description": "Z 座標"}
                    },
                    "required": ["x", "y", "z"]
                }
            }
        ]
    }
最後 AI 會根據 使用者的 Prompt 來決定要使用哪個工具
然後會自己調用 tools/call 這個地方
這裡會直接 send_unity_request , 發送指令到 Unity 的遊戲中
然後等待回復, Unity 回復之後, 就把他列印出來
elif method == "tools/call":
        tool_name = params.get("name")
        arguments = params.get("arguments", {})
        
        log(f"Calling tool: {tool_name} with arguments: {arguments}")
        
        if tool_name == "get_player_position":
            result = await send_unity_request("get_position")
            if "error" in result:
                return {"error": result["error"]}
            return {
                "content": [
                    {
                        "type": "text",
                        "text": f"玩家當前位置:\nX: {result.get('x', 0):.2f}\nY: {result.get('y', 0):.2f}\nZ: {result.get('z', 0):.2f}"
                    }
                ]
            }
        
        elif tool_name == "move_player":
            x = arguments.get("x")
            y = arguments.get("y")
            z = arguments.get("z")
            
            if x is None or y is None or z is None:
                return {"error": "Missing required parameters: x, y, z"}
            
            result = await send_unity_request("move_player", x=x, y=y, z=z)
            if "error" in result:
                return {"error": result["error"]}
            
            return {
                "content": [
                    {
                        "type": "text",
                        "text": f"✓ 玩家已移動到:\nX: {result.get('x', 0):.2f}\nY: {result.get('y', 0):.2f}\nZ: {result.get('z', 0):.2f}"
                    }
                ]
            }
基於這點, 我們就可以無限的擴展我們的 MCP 遊戲工具了
那麼架設好之後, 我們要在想要使用的AI工具上設定 MCP Config
以 Cursor 舉例, 你要到 Cursor 的MCP.json 中加入執行 Python 代碼的指令
{
  "mcpServers": {
    "unity-game-mcp": {
      "command": "python",
      "args": [
        "Assets/SimpleMCP/PythonServer/simple_mcp_bridge.py"
      ],
      "env": {
        "UNITY_HOST": "localhost",
        "UNITY_PORT": "8765"
      },
      "enabled": true
    }
  }
}
成功的話就會得到這樣的畫面:

Claude 或是別的AI工具也是一樣, 可以到對應的 Config 文件上加上這個 MCP Config 資訊
遊戲測試:
一切都准備好之後, 你可以直接在 Unity Editor 運行遊戲進行測試
或是 Build 出來進行, 我測試過, 兩種方式都是可以運行的
測試方法是先運行 Unity, 再啟動 MCP
然後就讓 AI 來操作遊戲角色:
Build Demo:
以上就是一個簡單的 In Game MCP 效果的實現
你可以繼續延伸擴展更多的應用 :)